// chat_server.cpp
#include "chat_server.h"
#include "timer_manager.h"
#include <sys/socket.h>
#include <arpa/inet.h>
#include <unistd.h>
#include <iostream>

ChatServer::ChatServer(const std::string& ip, int port, int threadCount, MySQLConnectionPool& pool)
    : m_stop(false), m_sqlHandler(pool),
    timer_fetch(std::chrono::seconds(20)),
      timer_proc(std::chrono::seconds(10)),
      timer_online(std::chrono::seconds(20)) {
    // 创建套接字
    m_sfd = socket(AF_INET, SOCK_STREAM, 0);
    if (m_sfd < 0) {
        std::cerr << "Failed to create socket" << std::endl;
        return;
    }
    
    // 设置套接字选项
    int opt = 1;
    if (setsockopt(m_sfd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof(opt)) < 0) {
        std::cerr << "Failed to set socket options" << std::endl;
        return;
    }
    
    // 绑定地址
    sockaddr_in serverAddr;
    serverAddr.sin_family = AF_INET;
    serverAddr.sin_port = htons(port);
    serverAddr.sin_addr.s_addr = inet_addr(ip.c_str());
    
    if (bind(m_sfd, (sockaddr*)&serverAddr, sizeof(serverAddr)) < 0) {
        std::cerr << "Failed to bind socket" << std::endl;
        return;
    }
    
    // 监听连接
    if (listen(m_sfd, 10) < 0) {
        std::cerr << "Failed to listen on socket" << std::endl;
        return;
    }
    
     initThreadPool(threadCount);

   timer_fetch.registerCallback([this]() {

        if (m_stop) {
            return; // 服务端已停止，不提交新任务
        }
        enqueueTask([this]() { this->FetchMsg(); });
        enqueueTask([this]() { this->FetchReq(); });
    });
    
    timer_proc.registerCallback([this]() {
      
        if (m_stop) {
            return; // 服务端已停止，不提交新任务
        }
        enqueueTask([this]() { this->SendMsg(); });
        enqueueTask([this]() { this->SendReq(); });
        enqueueTask([this]() { this->MoveToTemp();});
    });
    
    timer_online.registerCallback([this]() {
  
        if (m_stop) {
            return; // 服务端已停止，不提交新任务
        }
        enqueueTask([this]() { this->OnlineMsg(); });
        enqueueTask([this]() { this->OnlineReq(); });
    });

}

ChatServer::~ChatServer() {
    stop();
    close(m_sfd);
}

void ChatServer::initThreadPool(int threadCount) {
    for (int i = 0; i < threadCount; ++i) {
        m_workers.emplace_back([this]() {
            while (true) {
                std::function<void()> task;
                {
                    std::unique_lock<std::mutex> lock(this->m_taskMutex);
                    // 等待任务或停止信号
                    this->m_taskCV.wait(lock, [this]() {
                        return this->m_stop || !this->m_tasks.empty();
                    });  
                    // 如果停止且没有任务，退出线程
                    if (this->m_stop && this->m_tasks.empty())
                        return;
                    
                    // 取出任务
                    task = std::move(this->m_tasks.front());
                    this->m_tasks.pop();
                }
                
                // 执行任务
                task();
            }
        });
    }
}

// 添加任务到队列
void ChatServer::enqueueTask(std::function<void()> task) {
    {
        std::lock_guard<std::mutex> lock(m_taskMutex);
        if (m_stop||m_tasks.size()>1000) 
        {
            std::cerr << "任务队列已满或服务端已停止，拒绝新任务" << std::endl;
            return;  // 已停止则不添加任务
        }
        m_tasks.emplace(task);
        // std::cout << "任务入队，当前队列大小: " << m_tasks.size() << std::endl;
        // std::cout << "连接数: " << MySQLConnectionPool::getInstance().getCurrentSize() << std::endl;
    }
    m_taskCV.notify_one();  // 通知一个等待的线程
}

void ChatServer::start() {
    m_stop=false;

    timer_fetch.start();
    timer_proc.start();
    timer_online.start();

    std::cout << "Server started. Listening on port " << ntohs(((sockaddr_in*)&m_sfd)->sin_port) << std::endl;
    
    while (!m_stop) {
        sockaddr_in clientAddr;
        socklen_t clientAddrLen = sizeof(clientAddr);
        int clientFd = accept(m_sfd, (sockaddr*)&clientAddr, &clientAddrLen);
        
        if (clientFd < 0) {
            std::cerr << "Failed to accept client connection" << std::endl;
            continue;
        }
        std::cout<<"client_fd="<<clientFd<<std::endl;
        // 将客户端连接添加到任务队列
         enqueueTask([this, clientFd]() {
            handleClient(clientFd);  // 处理客户端连接的逻辑不变
        });
        
    }
}

void ChatServer::stop() {
    timer_fetch.stop();
    timer_proc.stop();
    timer_online.stop();
    
    {
        std::lock_guard<std::mutex> lock(m_taskMutex);
        m_stop = true;
    }
    m_taskCV.notify_all();
    for (auto& worker : m_workers) {
        if (worker.joinable())
            worker.join();
    }
}
// 检测是否为恶意协议请求
bool ChatServer::detectMaliciousProtocol(const std::string& message, const std::string& clientIP) {
    // 1. 检测常见攻击协议特征
    for (const auto& protocol : maliciousProtocols) {
        if (message.find(protocol) == 0) {
            std::cout << "检测到疑似" << protocol << "协议请求 from " << clientIP << std::endl;
            return true;
        }
    }
    
    // 2. 检测敏感路径访问
    if (message.find("HTTP/") != std::string::npos) {
        size_t firstLineEnd = message.find("\r\n");
        if (firstLineEnd != std::string::npos) {
            std::string firstLine = message.substr(0, firstLineEnd);
            for (const auto& path : sensitivePaths) {
                if (firstLine.find(path) != std::string::npos) {
                    std::cout << "检测到敏感路径访问: " << path << " from " << clientIP << std::endl;
                    return true;
                }
            }
        }
    }
    
    // 3. 检测恶意User-Agent
    size_t userAgentPos = message.find("User-Agent: ");
    if (userAgentPos != std::string::npos) {
        size_t agentEnd = message.find("\r\n", userAgentPos);
        if (agentEnd != std::string::npos) {
            std::string agent = message.substr(userAgentPos + 12, agentEnd - userAgentPos - 12);
            for (const auto& badAgent : maliciousUserAgents) {
                if (agent.find(badAgent) != std::string::npos) {
                    std::cout << "检测到恶意User-Agent: " << agent << " from " << clientIP << std::endl;
                    return true;
                }
            }
        }
    }
    
    return false;
}

bool ChatServer::detectAbnormalFrequency(const std::string& clientIP) {
    if (blockedIPs.find(clientIP) != blockedIPs.end()) {
        std::cout << "IP已被禁止: " << clientIP << std::endl;
        return true;
    }
    
    auto& times = ipAccessTimes[clientIP];
    std::time_t now = std::time(nullptr);
    
    // 清理过期记录（超出时间窗口）
    times.erase(std::remove_if(times.begin(), times.end(), 
    [this, now](std::time_t t) { return now - t > this->TIME_WINDOW; }), times.end());
    
    // 添加当前访问时间
    times.push_back(now);
    
    // 检测频率是否超标
    if (times.size() > MAX_ACCESS_FREQ) {
        std::cout << "IP访问频率异常: " << clientIP << ", 次数: " << times.size() << std::endl;
        blockedIPs.insert(clientIP); // 加入禁止列表
        return true;
    }
    
    return false;
}

void ChatServer::handleBlockedRequest(int clientFd, const std::string& clientIP, const std::string& reason) {
    std::string response = "BLOCKED|" + reason;
    send(clientFd, response.c_str(), response.length(), 0);
    std::cout << "拦截请求 from " << clientIP << ": " << reason << std::endl;
    
    // 关闭连接
    if (shutdown(clientFd, SHUT_WR) < 0) {
        std::cerr << "关闭写通道失败" << std::endl;
    }
    close(clientFd);
}

// 检测SQL注入（简化版）
bool ChatServer::detectSqlInjection(const std::string& message) {
    static const std::unordered_set<std::string> sqlKeyWords = {
        "SELECT ", "INSERT ", "UPDATE ", "DELETE ", "DROP ", "TRUNCATE ",
        "UNION ", "OR ", "AND ", "LIKE ", "%%", "'", "\"", "--", "/*", "*/"
    };
    
    for (const auto& keyword : sqlKeyWords) {
        if (message.find(keyword) != std::string::npos) {
            return true;
        }
    }
    return false;
}

// 处理客户端消息（在ChatServer::handleClient中）
void ChatServer::handleClient(int clientFd) {
    char buffer[1024];
    std::cout << "客户端连接已建立: " << clientFd << std::endl;
    
    // 关键：设置 clientFd 为非阻塞模式（就在这里设置）
    int flags = fcntl(clientFd, F_GETFL, 0);  // 获取当前文件状态标志
    if (flags == -1) {
        std::cerr << "fcntl 获取标志失败: " << strerror(errno) << std::endl;
        close(clientFd);
        return;
    }
    if (fcntl(clientFd, F_SETFL, flags | O_NONBLOCK) == -1) {  // 添加非阻塞标志
        std::cerr << "fcntl 设置非阻塞失败: " << strerror(errno) << std::endl;
        close(clientFd);
        return;
    }

    // 获取客户端IP
    sockaddr_in clientAddr;
    socklen_t clientAddrLen = sizeof(clientAddr);
    if (getpeername(clientFd, (sockaddr*)&clientAddr, &clientAddrLen) < 0) {
        std::cerr << "获取客户端IP失败" << std::endl;
        close(clientFd);
        return;
    }
    std::string clientIP = inet_ntoa(clientAddr.sin_addr);
    std::cout << "客户端IP: " << clientIP << std::endl;
    
    // 记录连接活跃时间
    std::time_t lastActiveTime = std::time(nullptr);
    const int TIMEOUT_SECONDS = 300;  // 5分钟超时

    while (!m_stop) {
        // 清空缓冲区
        memset(buffer, 0, sizeof(buffer));
        ssize_t bytesRead = recv(clientFd, buffer, sizeof(buffer), 0);
        
        if (bytesRead > 0) {
            // 更新活跃时间
            lastActiveTime = std::time(nullptr);
            std::string message(buffer, bytesRead);
            std::cout << "接收到消息: " << message << std::endl;
            
            // 1. 检测恶意协议
            if (detectMaliciousProtocol(message, clientIP)) {
                handleBlockedRequest(clientFd, clientIP, "恶意协议请求");
                break;
            }
            
            // 2. 检测异常访问频率
            if (detectAbnormalFrequency(clientIP)) {
                handleBlockedRequest(clientFd, clientIP, "访问频率过高");
                break;
            }
            
            // 3. 处理正常业务逻辑
            size_t delimiterPos = message.find('|');
            if (delimiterPos != std::string::npos) {
                std::string funcName = message.substr(0, delimiterPos);
                std::string paramsStr = message.substr(delimiterPos + 1);
                
                // 分割参数
                std::vector<std::string> params;
                size_t pos = 0,start_pos=0;
                std::cout<<"paramsStr="<<paramsStr<<std::endl;
                while ((pos = paramsStr.find('|',start_pos)) != std::string::npos) {
                    params.push_back(paramsStr.substr(start_pos, pos-start_pos));
                    start_pos=pos+1;
                } 
                if (!paramsStr.empty()) {
                    params.push_back(paramsStr.substr(start_pos));
                }
                
                // 执行SQL操作
                std::string result="";
                bool success = m_sqlHandler.execute(funcName, params, result);
                if(funcName=="query_login_1")
                {
                    if(result.size()>=3)
                    result.erase(result.size()-3);
                    if(result.length()>21)
                    {
                        {
                        std::lock_guard<std::mutex> lock(m_id_fd);
                        id_fd[std::stoll(params[0])]=clientFd;
                        }
                        std::string res_tmp="";
                        bool t_success=m_sqlHandler.execute("update_login",std::vector<std::string>{"1",params[0]},res_tmp);
                        if(t_success)
                        {
                            std::cout<<"login user_id="<<params[0]<<std::endl;
                            q_user_ol.emplace_back(std::stoll(params[0]));
                        }
                    }
                }else if(funcName=="insert_message")
                {
                    // if(result=="insert_message|success")
                    // {
                    //     std::string t_result="";
                    //     std::vector<std::string> t_params={params[0],params[1]};
                    //     long long mem_id=0;
                    //     bool t_success=m_sqlHandler.execute("hasRequest",t_params,t_result);
                    //     if(t_success)
                    //     {
                    //         if(t_result.length()<11)
                    //         {
                    //             std::reverse(t_params.begin(),t_params.end());
                    //             t_result="";
                    //             bool s_success=m_sqlHandler.execute("hasRequest",t_params,t_result);
                    //             size_t s_pos=t_result.find("|");
                    //             ++s_pos;
                    //             size_t e_pos=t_result.find(";;;");
                    //             mem_id=std::stoll(t_result.substr(s_pos,e_pos-s_pos));
                    //         }
                    //         else  
                    //         {
                    //             size_t s_pos=t_result.find("|");
                    //             ++s_pos;
                    //             size_t e_pos=t_result.find(";;;");
                    //             mem_id=std::stoll(t_result.substr(s_pos,e_pos-s_pos));
                    //         }

                    //         t_result="";
                    //         std::vector<std::string> v_mem;
                    //         v_mem.emplace_back(std::to_string(mem_id));
                    //         bool u_success=m_sqlHandler.execute("update_count",v_mem,t_result);
                    //          if(u_success)
                    //          {
                    //             if(t_result=="update_count|success")
                    //             {
                    //                 std::cout<<"update_count success"<<std::endl;
                    //             }
                    //          }
                    //     }
                    // }
                }
                {    
                    // 返回结果给客户端
                    std::cout << "操作了，结果: " << result << std::endl;
                    if(result.find(";;;")!=std::string::npos)
                    {
                        size_t st_pos=0,t_pos=0;
                        st_pos=result.find('|');
                        std::string tmp_func=result.substr(0,st_pos);
                        while((t_pos=result.find(";;;",st_pos))!=std::string::npos)
                        {
                            std::string tmp=tmp_func+result.substr(st_pos,t_pos-st_pos);
                            st_pos=t_pos+1;
                            if(send(clientFd,tmp.c_str(),tmp.length(),0)<0)
                            {
                                std::cerr<<"发送响应失败"<<strerror(errno)<<std::endl;
                                break;
                            }
                        }

                        if(funcName=="close_login")
                        {
                            {
                                std::lock_guard<std::mutex> lock(m_id_fd);
                                id_fd.erase(std::stoll(params[0]));
                                // 验证清理结果
                                if (id_fd.count(std::stoll(params[0])) > 0) {
                                    std::cerr << "id_fd 清理失败" << std::endl;
                                }
                            }
                            std::cout<<"close_login success"<<std::endl;
                            break;
                        }


                    }else 
                    {   if (send(clientFd, result.c_str(), result.length(), 0) < 0) {
                        std::cerr << "发送响应失败: " << strerror(errno) << std::endl;
                        break;
                    }
                }
            }
            }
        } else if (bytesRead == 0) {
            // 客户端主动关闭连接
            std::cout << "客户端正常断开连接" << std::endl;
            break;
        } else {
            // 处理recv错误
            if (errno == EAGAIN || errno == EWOULDBLOCK) {
                // 非阻塞模式下无数据可读，不是错误，继续等待
                // 短暂休眠避免CPU忙轮询
                std::this_thread::sleep_for(std::chrono::milliseconds(10));
            } else {
                // 真正的错误（如连接重置、被关闭等）
                std::cerr << "接收数据失败（真正错误）: " << strerror(errno) << std::endl;
                break;
            }
        }
        
        // 检查超时
        if (std::time(nullptr) - lastActiveTime > TIMEOUT_SECONDS) {
            std::cout << "客户端超时断开: " << clientIP << std::endl;
            break;
        }
    }
    
    // 关闭连接
    close(clientFd);
    std::cout << "客户端连接已关闭: " << clientFd << std::endl;
}

size_t ChatServer::FindNthStr(const std::string& source,const std::string& goal,int nth)
{
    if(nth<=0||goal.empty()) return std::string::npos;
    size_t pos=0;
    for(int i=0;i<nth;++i)
    {
        pos=source.find(goal,pos);
        if(pos == std::string::npos) return std::string::npos;
        if(i!=nth-1) pos+=1;
    }
    return pos;
}

void ChatServer::FetchMsg()
{
    std::string result;
    std::vector<std::string> params;
    bool success=m_sqlHandler.execute("getMessages",params,result);
    if(success)
    {
        std::cout<<"fetch msg"<<std::endl;
        size_t s_pos=0,pos=0;
        s_pos=result.find("|");
        if(s_pos==std::string::npos||result.empty())
        {
            std::cout<<"fetch no msg"<<std::endl;
            return ;
        }
        ++s_pos;//移动位置
        std::vector<long long> tmp;
        if(result!="getMessages")
        {
            while((pos=result.find(";;;",s_pos))!=std::string::npos)
            {
                tmp.emplace_back(std::stoll(result.substr(s_pos,pos-s_pos)));
                s_pos=pos+4;
            }
        }
        {
        std::lock_guard<std::mutex> lock(m_msg_life);
        online_msg.insert(make_move_iterator(tmp.begin()),
                        make_move_iterator(tmp.end()));
        std::cout<<"msg_life = "<<online_msg.size()<<std::endl;
        }

        
    }
    // m_msg_avail=true;
    // m_msg_cv.notify_one();

}
void ChatServer::FetchReq()
{
    std::string result="";
    std::vector<std::string> params;
    bool success=m_sqlHandler.execute("getRequests",params,result);
    if(success)
    {
        std::cout<<"fetch req"<<std::endl;
        size_t s_pos=0,pos=0;
        s_pos=result.find("|");
        if(s_pos==std::string::npos)
        {
            std::cout<<"fetch no req"<<std::endl;
            return ;
        }
        ++s_pos;
        std::vector<long long> tmp;
        if(result!="getRequests")
        {
            while((pos=result.find(";;;",s_pos))!=std::string::npos)
            {
                tmp.emplace_back(std::stoll(result.substr(s_pos,pos-s_pos)));
                s_pos=pos+4;
            }
        }
        {
            std::lock_guard<std::mutex> lock(m_req_life);
            online_req.insert(make_move_iterator(tmp.begin()),   
                            make_move_iterator(tmp.end()));
        }
    }
}
void ChatServer::SendMsg()
{
    std::unordered_set<long long> tmp;
    {
        std::lock_guard<std::mutex> lock(m_msg_life);
        tmp=std::move(online_msg);
    }
    std::string result="";
    std::vector<std::string> params;
    std::unordered_map<long long,std::vector<long long>> um_tmp;
    for(long long msg_id: tmp)
    {
        std::cout<<"msg_id="<<msg_id<<std::endl;
        result="";
        std::string str=std::to_string(msg_id);
        params.emplace_back(str);
        bool success=m_sqlHandler.execute("selectMessage",params,result);
        if (result.size() >= 3) {
        result.erase(result.size() - 3);
        }
        std::cout<<"result="<<result<<std::endl;
        if(success)
        {
            size_t f_pos=FindNthStr(result,"|",2);
            size_t t_pos=FindNthStr(result,"|",3);
            long long user_id=std::stoll(result.substr(f_pos+1,t_pos-f_pos-1));
            std::cout<<"user_id="<<user_id<<std::endl;

            {   
                std::lock_guard<std::mutex> lock(m_id_fd);
                if(id_fd.find(user_id)!=id_fd.end())
                {
                    int recv_id=id_fd[user_id];

                    if (fcntl(recv_id, F_GETFL) == -1) {
                        std::cerr << "无效的 clientFd=" << recv_id << std::endl;
                        continue;
                    }

                    // std::cout<<"recv_id="<<recv_id<<std::endl;
                    ssize_t bytes_sent = send(recv_id, result.c_str(), result.length(), 0);

                    if (bytes_sent == -1) {
                        if (errno == EAGAIN || errno == EWOULDBLOCK) {
                            std::cout << "SendMsg: 发送缓冲区满，稍后重试，msg_id=" << msg_id << std::endl;
                            um_tmp[user_id].emplace_back(msg_id);
                        } else {
                            std::cout << "SendMsg: 发送失败，错误: " << strerror(errno) << "，msg_id=" << msg_id << std::endl;
                            um_tmp[user_id].emplace_back(msg_id);
                        }
                    } else {
                        // std::cout << "SendMsg: 发送成功，发送字节数=" << bytes_sent << "，msg_id=" << msg_id << std::endl;
                        result="";
                                bool u_success=m_sqlHandler.execute("update_message",std::vector<std::string>{std::to_string(msg_id)},result);
                                if(u_success)
                                {
                                    if(result=="update_message|success")
                                    std::cout<<"update_message success"<<std::endl;
                                }
                        // 更新消息状态为已接收（update_message）
                    }

                    params.clear();
                }else std::cout<<"user_id can't found"<<std::endl;
            }
        }
    }
    {
        std::lock_guard<std::mutex> lock(m_msg_dead);   
        offline_msg.merge(std::move(um_tmp));
        // std::cout<<"offline_msg = "<<offline_msg.size()<<std::endl;
    }
}
void ChatServer::SendReq()
{
    std::unordered_set<long long> tmp;
    {
        std::lock_guard<std::mutex> lock(m_req_life);
        tmp=std::move(online_req);
    }
    std::string result="";
    std::vector<std::string> params;
    std::unordered_map<long long,std::vector<long long>> um_tmp;
    for(long long req_id: tmp)
    {
        result="";
        std::string str=std::to_string(req_id);
        params.emplace_back(str);
        bool success=m_sqlHandler.execute("selectRequest",params,result);
        if (result.size() >= 3) {
        result.erase(result.size() - 3);
        }
        if(success)
        {
            size_t f_pos=FindNthStr(result,"|",3);
            size_t t_pos=FindNthStr(result,"|",4);
            long long user_id=std::stoll(result.substr(f_pos+1,t_pos-f_pos-1));
            {   
                std::lock_guard<std::mutex> lock(m_id_fd);
                if(id_fd.find(user_id)!=id_fd.end())
                {
                    int recv_id=id_fd[user_id];
                    if(send(recv_id,result.c_str(),result.length(),0)<0)
                    {
                        std::cerr<<"发送响应失败"<<std::endl;
                        um_tmp[user_id].emplace_back(req_id);
                    }else {
                        result="";
                        bool u_success=m_sqlHandler.execute("update_request_1",std::vector<std::string>{std::to_string(req_id)},result);
                        if(u_success)
                        {
                            if(result=="update_request_1|success")
                            std::cout<<"udpate_request_1 success"<<std::endl;
                        }
                    }
                }else um_tmp[user_id].emplace_back(req_id);
            }    
        }
        params.clear();
    }

    {
        std::lock_guard<std::mutex> lock(m_req_dead);   
        offline_req.merge(std::move(um_tmp));
    }
}
void ChatServer::OnlineMsg()
{
    
    std::vector<long long> temp;//临时容器来防止嵌套锁导致死锁
    {
        std::lock_guard<std::mutex> lock(m_msg_tmp);
        temp=std::move(q_msg_tmp);
    }

    {
        std::lock_guard<std::mutex> lock(m_msg_life);
        online_msg.insert(make_move_iterator(temp.begin()),
                        make_move_iterator(temp.end()));
        // std::cout<<"online_msg ol = "<<online_msg.size()<<std::endl;
    }
    
}
void ChatServer::OnlineReq()
{
    std::vector<long long> temp;
    {
        std::lock_guard<std::mutex> lock(m_req_tmp);
        temp=std::move(q_req_tmp);
    }

    {
        std::lock_guard<std::mutex> lock(m_req_life);
        online_req.insert(make_move_iterator(temp.begin()),
                        make_move_iterator(temp.end()));
    }
}

void ChatServer::MoveToTemp()
{
    bool has_data=false;
    std::vector<long long> c_user_temp; 
    {
        std::unique_lock<std::mutex> lock(m_user_ol);
        c_user_temp=std::move(q_user_ol);
        if(!c_user_temp.empty())
            has_data=true;
    }

    if(has_data)
    {
        for(long long user_id: c_user_temp)
        {
            std::vector<long long> msg_ids;
            {
                std::lock_guard<std::mutex> lock(m_msg_dead);
                auto it=offline_msg.find(user_id);
                if(it!=offline_msg.end())
                {
                    msg_ids=std::move(it->second);
                    offline_msg.erase(it);
                }
                
                if(!msg_ids.empty())
                {
                    std::lock_guard<std::mutex> lock(m_msg_tmp);
                    q_msg_tmp.insert(q_msg_tmp.end(),
                                    std::make_move_iterator(msg_ids.begin()),
                                    std::make_move_iterator(msg_ids.end()));
                    // std::cout<<"q_msg_tmp = "<<q_msg_tmp.size()<<std::endl;
                }
            }

            std::vector<long long> req_ids;
            {
                std::lock_guard<std::mutex> lock(m_req_dead);
                auto it=offline_req.find(user_id);
                if(it!=offline_req.end())
                {
                    req_ids=std::move(it->second);
                    offline_req.erase(it);
                }
                
                if(!req_ids.empty())
                {
                    std::lock_guard<std::mutex> lock(m_req_tmp);
                    q_req_tmp.insert(q_req_tmp.end(),
                                    std::make_move_iterator(req_ids.begin()),
                                    std::make_move_iterator(req_ids.end()));
                }
            }
        }
    }
    

}